package HandshakeJoin;

import org.apache.s4.core.ProcessingElement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ResultPersistor extends ProcessingElement
{
    transient private static final Logger logger = LoggerFactory
            .getLogger(ResultPersistor.class);

    private int numFields = 7;
    private boolean consoleDisplay = true;
    private String dbChoice;
    private DBClient db;
    private long recordsCount;

    @Override
    protected void onCreate() {
        db = null;
        if (dbChoice != null) {
            if (dbChoice.equals("cassandra")) {
                db = new CassandraClient("join" + getId(), numFields);
            }
            if (dbChoice.equals("hbase")) {
                db = new HBaseClient("join" + getId());
            }
            else if (dbChoice.equals("logbase")) {
                db = new LogBaseClient("join" + getId(), numFields);
            }
        }

        if (db != null) {
            db.init();
        }

        recordsCount = 0;
    }

    public void onEvent(Snapshot snapshot) {
        int numRecords = snapshot.getRecordsCount();
        recordsCount += numRecords;

        if (consoleDisplay && snapshot != null) {
            System.out.println(snapshot);
        }
        System.out.println("Results of join " + getId() + ": " + numRecords + "/" + recordsCount);

        materialize(snapshot);
    }

    @Override
    protected void onRemove() {
        if (db != null) {
            db.close();
        }
    }

    private void materialize(final Snapshot snapshot) {
        if (db != null) {
            db.materialize(snapshot);
        }
    }

    public ResultPersistor setConsoleDisplay(final boolean consoleDisplay) {
        this.consoleDisplay = consoleDisplay;

        return this;
    }

    public ResultPersistor setNumberOfFields(final int numFields) {
        this.numFields = numFields;

        return this;
    }

    public ResultPersistor setDBChoice(final String dbChoice) {
        this.dbChoice = dbChoice;

        return this;
    }
}